package com.itqiqi.api.tableapi;

import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.java.BatchTableEnvironment;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.types.Row;

public class TableTest2_commonApi {

    public static void main(String[] args) throws Exception {

        // TODO: 2022/5/18 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // TODO: 2022/5/18 创建表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO: 2022/5/18 基于老版本planner的流处理
        EnvironmentSettings oldStreamSettings = EnvironmentSettings.newInstance()
                .useOldPlanner()
                .inStreamingMode()
                .build();
        StreamTableEnvironment streamTableEnv = StreamTableEnvironment.create(env, oldStreamSettings);

        // TODO: 2022/5/18 基于老版本planner的批处理
        ExecutionEnvironment batchEnv = ExecutionEnvironment.getExecutionEnvironment();
        BatchTableEnvironment batchTableEnv = BatchTableEnvironment.create(batchEnv);

        // TODO: 2022/5/18 基于Blink的流处理
        EnvironmentSettings blinkStreamSettings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        StreamTableEnvironment blinkStreamTableEnv = StreamTableEnvironment.create(env, blinkStreamSettings);

        // TODO: 2022/5/18 基于Blink的批处理
        EnvironmentSettings blinkBatchSettings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inBatchMode()
                .build();
        TableEnvironment blinkBatchTableEnv = TableEnvironment.create(blinkBatchSettings);

        // TODO: 2022/5/18 连接外部文件系统
        tableEnv.connect( new FileSystem().path("input/sensor.txt"))
                .withFormat( new Csv())
                .withSchema( new Schema()
                        .field("id", DataTypes.STRING())
                        .field("timestamp", DataTypes.BIGINT())
                        .field("temp", DataTypes.DOUBLE())
                )
                .createTemporaryTable("inputTable");

        Table inputTable = tableEnv.from("inputTable");
        inputTable.printSchema();
        tableEnv.toAppendStream(inputTable, Row.class).print("inputTable");


        // TODO: 2022/5/18 查询转换 TAble API
        // 简单转化
        Table resTable = inputTable.select("id, temp")
                .filter("id = 'sensor_1'");

        // 聚合统计
        Table aggTable = inputTable.groupBy("id")
                .select("id, id.count as cnt, temp.avg as avgTemp");

        // TODO: 2022/5/18 SQl
        tableEnv.sqlQuery("select id, temp from inputTable where id = 'sensor_1'");
        Table sqlTable = tableEnv.sqlQuery("select id, count(id) as cnt, avg(temp) as avgTemp from inputTable group by id");

        tableEnv.toAppendStream(resTable, Row.class).print("resTable");
        tableEnv.toRetractStream(aggTable, Row.class).print("aggTable");
        tableEnv.toRetractStream(sqlTable, Row.class).print("sqlTable");

        env.execute();

    }
}
